package com.echo.code.thread.pool.impl;

import com.echo.code.thread.pool.ThreadPool;
import com.echo.code.thread.queue.BlockingQueue;
import com.echo.code.thread.queue.impl.LinkedBlockingQueue;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/**
 * 线程池默认实现
 */
public class DefaultThreadPool extends Thread implements ThreadPool {

  // jdk日志记录器，不依赖于其它框架
  public static final Logger LOGGER = Logger.getLogger("com.echo.code.thread.pool.impl.DefaultThreadPool");

  // 线程池初始线程数
  private final int initSize;

  // 线程池核心线程数
  private final int coreSize;

  // 线程池最大线程数
  private final int maxSize;

  // 线程池当前活跃的线程数
  private int activeCount;

  // 用于存放线程任务的队列
  private final BlockingQueue<Runnable> blockingQueue;

  // 线程池空闲线程存活时间(单位：毫秒)
  private final long keepAliveTime;

  // 用于存放工作线程的队列
  private final Queue<ThreadTask> threadQueue = new ArrayDeque<>();

  // private final ThreadFactory threadFactory = null; TODO 创建工作线程所需要的工厂,为了保持线程池的实现简单，暂不实现

  // 线程池是否关闭
  private volatile boolean isShutdown = false;

  // 构造时需要传递的参数：初始线程数, 最大线程数, 核心线程数
  public DefaultThreadPool(int initSize, int maxSize, int coreSize) {
    this.initSize = initSize;
    this.maxSize = maxSize;
    this.coreSize = coreSize;
    this.keepAliveTime = 20000;
    this.blockingQueue = new LinkedBlockingQueue<>();
    this.init();
    LOGGER.info("============= 线程池初始化成功，初始化线程数:" + this.initSize + " =============");
  }

  // 构造时需要传递参数, 该构造函数需要参数比较多
  public DefaultThreadPool(int initSize, int maxSize, int coreSize, int queueSize, long keepAliveTime) {
    this.initSize = initSize;
    this.maxSize = maxSize;
    this.coreSize = coreSize;
    this.blockingQueue = new LinkedBlockingQueue<>(queueSize);
    this.keepAliveTime = keepAliveTime;
    this.init();
    LOGGER.info("============= 线程池初始化成功，初始化线程数:" + this.initSize + " =============");
  }

  // 线程池初始化
  private void init() {
    for (int i = 0; i < initSize; i++) {
      newThread();
    }
    start();
  }

  @Override
  public void run() {
    while (!isShutdown) {
      try {
        TimeUnit.MILLISECONDS.sleep(keepAliveTime);
      } catch (InterruptedException e) {
        e.printStackTrace();
        // TODO 可添加中断逻辑
      }
      synchronized (this) {
        if (isShutdown) {
          break;
        }
        // 当前有任务尚未处理, 且activeCount < coreSize则继续扩容
        if (blockingQueue.size() > 0 && activeCount < coreSize) {
          for (int i = initSize; i < coreSize; i++) {
            newThread();
          }
          // continue的目的在于不想让线程的扩容直接达到maxsize
          continue;
        }
        // 当前有任务尚未处理, 且activeCount < maxSize则继续扩容
        if (blockingQueue.size() > 0 && activeCount < maxSize) {
          for (int i = coreSize; i < maxSize; i++) {
            newThread();
          }
        }
        // 如果没有任务, 则需要回收, 回收至coreSize即可
        if (blockingQueue.size() == 0 && activeCount > coreSize) {
          for (int i = coreSize; i < activeCount; i++) {
            removeThread();
          }
        }
      }
    }
  }

  @Override
  public void shutdown() {
    synchronized (this) {
      if (isShutdown) {
        return;
      }
      threadQueue.forEach(threadTask -> {
        threadTask.task.stop();
      });
      isShutdown = true;
      LOGGER.info("============= 线程池关闭 =============");
    }
  }

  @Override
  public int getActiveCount() {
    synchronized (this) {
      if (isShutdown) {
        activeCount = 0;
      }
      return activeCount;
    }
  }

  private void newThread() {
    // 创建任务线程, 并且启动
    Task task = new Task();
    Thread thread = new Thread(task);
    ThreadTask threadTask = new ThreadTask(thread, task);
    threadQueue.offer(threadTask);
    activeCount++;
    thread.start();
  }

  private void removeThread() {
    // 从线程池中移除某个线程
    ThreadTask threadTask = threadQueue.remove();
    threadTask.task.stop();
    activeCount--;
  }

  @Override
  public void execute(Runnable runnable) {
    if (isShutdown) {
      throw new IllegalStateException("The thread pool is destroy");
    }
    blockingQueue.offer(runnable);
  }

  @Override
  public int getInitSize() {
    if (isShutdown) {
      throw new IllegalStateException("The thread pool is destroy");
    }
    return initSize;
  }

  @Override
  public int getMaxSize() {
    if (isShutdown) {
      throw new IllegalStateException("The thread pool is destroy");
    }
    return maxSize;
  }

  @Override
  public int getCoreSize() {
    if (isShutdown) {
      throw new IllegalStateException("The thread pool is destroy");
    }
    return coreSize;
  }

  @Override
  public int getQueueSize() {
    if (isShutdown) {
      throw new IllegalStateException("The thread pool is destroy");
    }
    return blockingQueue.size();
  }

  @Override
  public boolean isShutdown() {
    return isShutdown;
  }

  /**
   * 内部任务类
   * */
  public class Task implements Runnable {

    // 任务开关
    private volatile boolean running = true;

    @Override
    public void run() {
      while (running) {
        Runnable task = blockingQueue.take();
        task.run();
      }
    }

    public void stop() {
      this.running = false;
    }
  }

  /**
   * 仅仅是将线程与任务结合
   * */
  private static class ThreadTask {

    public Thread thread;
    public Task task;

    public ThreadTask(Thread thread, Task task) {
      this.thread = thread;
      this.task = task;
    }
  }
}